package com.heima.article.stream;
import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.message.vos.ArticleVisitStreamMess;
import com.heima.model.message.vos.UpdateArticleMess;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

/**
 * 执行热点文章实时计算的类
 */
@Configuration
public class HotArticleStreamHandler {
    /**
     * 该方法表示实时计算的方法，交给IOC管理，服务启动的时候就会加载
     * @return
     */
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //1.接收消息，此时此刻就是消费者的角色
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //2.处理流式计算
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 进行接收消息，然后进行处理
             * @param key 表示发送消息的key  =null
             * @param value 表示消息的内容，字符串序列化之后的UpdateArticleMess
             * @return
             *   重新处理之后的key=文章id   value=行为的类型:行为的数量，比如点赞likes:1 取消点赞likes:-1
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1.把消息内容转化对象
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                return new KeyValue<>(mess.getArticleId().toString(),mess.getType()+":"+mess.getAdd());
            }
        })
                /**
                 * 进行分组处理
                 * key表示 文章id
                 * value表示行为的类型:行为的数量，比如点赞likes:1 取消点赞likes:-1
                 */
                .groupBy((key,value)->key)
                /**
                 * 事件窗口，每10s处理一次消息
                 */
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                /**
                 * 真正的处理流式计算的业务
                 */
                .aggregate(new Initializer<String>() {
                    /**
                     *初始化阶段,只会执行一次
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                },
                        new Aggregator<String, String, String>() {
                            /**
                             * 核心聚合方法
                             * @param key  表示文章id
                             * @param value  表示行为的类型:行为的数量，比如点赞likes:1 取消点赞likes:-1
                             * @param init   COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                             * @return  返回值是一个字符串，但是字符串的内容"COLLECTION:5,COMMENT:3,LIKES:1,VIEWS:2"
                             */
                            @Override
                            public String apply(String key, String value, String init) {
                                //1.判断value是否为空
                                if(StringUtils.isBlank(value)){
                                    return init;
                                }
                                //定义变量
                                int collection = 0,comment=0,like=0,view=0;
                                //对value进行字符串分割成数据
                                String[] split = value.split(":");
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                    case COLLECTION:
                                        collection+=Integer.parseInt(split[1]);
                                        break;
                                    case COMMENT:
                                        comment+=Integer.parseInt(split[1]);
                                        break;
                                    case VIEWS:
                                        view+=Integer.parseInt(split[1]);
                                        break;
                                    case LIKES:
                                        like+=Integer.parseInt(split[1]);
                                        break;
                                }

                                //拼接字符串
                                String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);

                                System.out.println("文章的id:"+key);
                                System.out.println("当前时间窗口内的消息处理结果："+format);
                                return format;
                            }
                },
                        ////表示聚合名称，可以随意起
                        Materialized.as("hot-article-name"))
                .toStream()
                /**
                 * 到此，能够走到当前这步的话，表示流式处理完成
                 * 该map方法表示封装数据流式处理之后的数据，整理要发送给消费者的消息内容
                 */
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    /**
                     * 封装处理之后的结果数据，用来重新发送消息
                     * @param key  发送消息的key  key.key() 转成字符串类型  表示的是文章id
                     * @param value 是"COLLECTION:5,COMMENT:3,LIKES:1,VIEWS:2"字符串
                     * @return 注意：必须保证发送的消息内容中要包含文章id以及上面的value内容
                     */
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key(), formatMessage(key.key(),value));
                    }
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return kStream;
    }

    /**
     * 封装消息的内容
     * @param key  文章id
     * @param value  是"COLLECTION:5,COMMENT:3,LIKES:1,VIEWS:2"字符串
     * @return 希望是一个对象，然后转化成json字符串作为消息内容
     */
    private String formatMessage(String key, String value) {
        //1.定义一个封装的消息对象
        ArticleVisitStreamMess mess=new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(key));

        String[] split = value.split(",");
        for (String s : split) {//每一个s表示：COLLECTION:5
            String[] split1 = s.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split1[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split1[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split1[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split1[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split1[1]));
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }
}
